﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Collections.Concurrent;

using WorkStation.Frameworks.Core;

namespace WorkStation.Frameworks.Core.Threading.Tasks
{
    /// <summary>
    /// 任务队列
    /// </summary>
    /// <typeparam name="TTask">任务类型</typeparam>
    public class TaskExecutorQueue<TTask> : IDisposable
    {
        protected Action<TTask, CancellationTokenSource> _action;
        protected Func<CancellationTokenSource, BlockingCollection<TTask>, bool> _stop;
        protected CancellationTokenSource _source = new CancellationTokenSource();
        protected Action<TTask> _callback;

        /// <summary>
        /// 队列
        /// </summary>
        public BlockingCollection<TTask> Queue { get; set; }

        /// <summary>
        /// 队列名称
        /// </summary>
        public string QueueName { get; set; }

        /// <summary>
        /// 线程
        /// </summary>
        public Task TaskObject { get; set; }

        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="name">队列名称，默认使用 Guid </param>
        public TaskExecutorQueue(string name = null) 
        {
            if (name == null) 
            {
                this.QueueName = Guid.NewGuid().ToString();
            }

            this.Queue = new BlockingCollection<TTask>();
        }

        /// <summary>
        /// 添加任务
        /// </summary>
        /// <param name="task">任务对象</param>
        /// <returns></returns>
        public TaskExecutorQueue<TTask> Push(TTask task) 
        {
            return this.Queue.Continue(v => v.Add(task)).Change(v => this);
        }

        /// <summary>
        /// 注册处理函数
        /// </summary>
        /// <param name="action">处理函数</param>
        /// <param name="stop">停止条件</param>
        /// <returns></returns>
        public TaskExecutorQueue<TTask> RegisterProcess(Action<TTask, CancellationTokenSource> action, Action<TTask> callback = null, Func<CancellationTokenSource, BlockingCollection<TTask>, bool> stop = null) 
        {
            this._action = action;
            this._stop = stop;
            this._callback = callback;

            return this;
        }

        /// <summary>
        /// 开始执行
        /// </summary>
        /// <returns></returns>
        public TaskExecutorQueue<TTask> Start() 
        {
            if (this._stop == null) { this._stop = (source, queue) => true; }

            bool isCallback = this._callback == null ? false : true;

            Action action = () =>
            {
                while (true)
                {
                    if (this._source.Token.IsCancellationRequested)
                    {
                        this._source.Token.ThrowIfCancellationRequested();
                    }

                    TTask taskInfo = this.Queue.Take(this._source.Token);

                    this._action(taskInfo, this._source);

                    if (isCallback) { this._callback(taskInfo); }
                }
            };

            this.TaskObject = new Task(action, this._source.Token, TaskCreationOptions.LongRunning);

            this.TaskObject.Start();

            return this;
        }

        /// <summary>
        /// 停止
        /// </summary>
        /// <exception cref="System.Exception">如果在处理过程中有异常，则在停止时抛出，推荐在处理任务时处理异常</exception>
        /// <returns></returns>
        public TaskExecutorQueue<TTask> Stop() 
        {
            while (!this._stop(this._source, this.Queue))
            {
                Thread.Sleep(300);
            }

            this._source.Cancel();

            try
            {
                this.TaskObject.Wait();
            }
            catch (AggregateException e)
            {
                if (!e.InnerExceptions.All(v => v is OperationCanceledException)) { throw e; }
            }

            return this;
        }

        /// <summary>
        /// 释放
        /// </summary>
        public void Dispose()
        {
            this.Stop();
        }
    }
}
